package org.apache.zookeeper.server.paxos2;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.BitSet;
import java.util.List;
import java.util.Map;
import java.util.LinkedList;

import org.apache.log4j.Logger;

/**
 * Created by IntelliJ IDEA. User: iodine Date: Jul 30, 2011 Time: 1:23:51 PM To
 * change this template use File | Settings | File Templates.
 */
public class Metadata {
	public static final String path = "/var/data/metadata";
	private static final Logger LOG = Logger.getLogger(Metadata.class);

	private BitSet validBits;
	private boolean validBitsDirty = false;

	private MetaStorage versions;

	public static byte[] longToBytes(long value) {
		byte[] ret = new byte[8];
		ByteBuffer b = ByteBuffer.wrap(ret);
		b.putLong(value);
		return ret;
	}

	public static long bytesToLong(byte[] value) {
		ByteBuffer b = ByteBuffer.wrap(value);
		return b.getLong();
	}

	public Metadata() {
		File f = new File(path + File.separator + "valid");
		if (f.exists()) {
			try {
				ObjectInputStream ois = new ObjectInputStream(
						new FileInputStream(f));
				validBits = (BitSet) (ois.readObject());
			} catch (Exception e) {
				e.printStackTrace();
				validBits = null;
			}
		}
		if(validBits == null){
            int totalBlocks = (int) (Configuration.totalSize / Configuration.blockSize);
            validBits = new BitSet(totalBlocks);

            boolean []belong = new boolean[(int)Configuration.noOfStripes];
            List<Integer> shards = Mapping.listShards(Configuration.serverId);
            for(Integer shard : shards){
		System.out.println("I have shard "+shard);
                belong[shard-1] = true;
            }
            for(int i=0; i<totalBlocks; i++){
                if(belong[i%(int)Configuration.noOfStripes]==true)
                    validBits.set(i);
            }
		}
		versions = new MetaStorage(path, "version");
	}

    public void clearValid(){
        this.validBits.clear();
    }

    public BlockMeta getBlockMeta(long blockId) {
        return new BlockMeta(ByteBuffer.wrap(versions.read(blockId)));
	}
	
	public void getBlockMeta(long blockId, BlockMeta blockMeta) {
		blockMeta.deserialize(ByteBuffer.wrap(versions.read(blockId)));
	}

	public void setBlockMeta(long blockId, BlockMeta meta) {
		versions.write(blockId, meta.toByteArray());
	}

	public boolean isValid(long blockId) {
		return validBits.get((int) blockId);
	}

	public void setValid(long blockId) {
		if (!validBits.get((int) blockId)) {
			validBitsDirty = true;
			validBits.set((int) blockId);
		}
	}

	public void setInvalid(long blockId) {
		if (validBits.get((int) blockId)) {
			validBitsDirty = true;
			validBits.clear((int) blockId);
		}
	}
	
	public void close(){
		this.versions.close();
	}

    private class NewBlockFinder implements MetaStorage.MetaFinder{
        private final long zxid;
        public NewBlockFinder(long zxid){
            this.zxid = zxid;
        }

        public boolean isOK(Map.Entry<Long, byte[]> entry) {
            //if(!isValid(entry.getKey()))
            //    return false;
            long version = bytesToLong(entry.getValue());
            if(version>zxid)
                return true;
            return false;
        }
    }

    public void getNewBlocks(long zxid, Storage storage, int sid){
        NewBlockFinder finder = new NewBlockFinder(zxid);
        versions.find(finder, storage, sid);
    }

    public void sync() {
		versions.flushFile();
		if (validBitsDirty) {
			try {
				ObjectOutputStream oos = new ObjectOutputStream(
						new FileOutputStream(path + File.separator + "valid"));
				oos.writeObject(validBits);
				oos.close();
				validBitsDirty = false;
			} catch (IOException e) {
				LOG.error("Could not sync meta", e);
				//e.printStackTrace();
				System.exit(-1);
			}
		}
	}
}
